[Day25] 架構優化 : Redis 實作 Server Cache

2023 iThome 鐵人賽

DAY 25

Modern Web

FastAPI 如何 Fast ？框架入門、實例、重構與測試系列第 25 篇

15th鐵人賽 python fastapi redis

zhu424

2023-10-10 22:58:57

923 瀏覽

分享至

[Day25] 架構優化 : Redis 實作 Server Cache

本次的程式碼與目錄結構可以參考 FastAPI Tutorial : Day25 branch

前言

在昨天我們完成 Redis 的基本操作包括：

建立 Redis 連線
- 使用 Redis 類別來建立連線
- 以 ConnectionPool 來管理連線
存取資料
- 以 redis-py 提供的 set 、 save
- 使用 redis-om 的 Object Mapping 來做 CRUD

有這些基本的認識後
今天可以嘗試使用 redis-om 來實作 Server Cache
並比較 fastapi-cache 和 redis-py 來實作 Cache 的差異

Server Cache 架構

這次想實作的 Server Cache 是在 CRUD 層面的 Cache
而不是在 Response Level 的 Cache

所以在遇到 Read 操作時，我們會先去 Cache 中尋找資料
如果 Cache 中有資料，則直接回傳
並把 user_id 當作 key 來存取

而 Create , Update , Delete 操作時
我們會依據 選定的 Key 來更新或刪除 key-value pair

如果使用 `fastapi-cache` 實作

fastapi-cache Github

因為該套件主要是針對 Response Level 的 Cache
是透過 decorator 來 Cache 我們要 Response 前的結果

用法大略如下：

from fastapi_cache.decorator import cache

@app.get("/users/{user_id}")
@cache(expire=60) # <----- 60 秒後過期
async def read_user(user_id: int):
    return {"user_id": user_id}

但是 fastapi-cache 無法做到遇到:
「 Create , Update , Delete 操作時，要依需相對應的 Key 來更新刪除 Key-value Pair 」的操作

如果使用 `redis-om` 實作

如果由 redis-om 來實作的話
我們也需要先定義 UserCacheModel

class UserCache( JsonModel ):
    id: Optional[int] = Field(index=True)
    name:Optional[str] = Field(index=True)
    password:Optional[str] = Field(index=False)
    name: Optional[str] = Field(index=True)
    avatar: Optional[str] = Field(index=False)
    age: Optional[int] = Field(index=False,default=0)
    email: Optional[EmailStr] = Field(index=True)
    birthday: Optional[date] = Field(index=False)

    class Meta:
        database = redis

為了盡量不改動到 CRUD 的 Code
我們可以使用 decorator 來實作
又因為讓 cache decorator 保留彈性，我們可以傳入 key 來當作 目前應該要依據哪個欄位搜尋
cls 則是 CRUD 回傳的 Schema

database/redis_cache.py

def generic_cache_get(key:str,cls):
    def inner(func):
        async def wrapper(*args, **kwargs):
            # TODO
            return wrapper
    return inner

再將內層 get 、 set 邏輯實作完
database/redis_cache.py

    try:
        redis_result = UserCache.find( ModelField(name=key,type=type(value_key)) == value_key ).first()
    except:
        redis_result = None

    if redis_result:
        return cls(**redis_result.dict())
    else:
        result = await func(*args, **kwargs)
        UserCache(**result.dict()).save()
        return result

使用 `redis-om` 內建 Model 實作的坑

會發現在使用上，每 Model.save() 一次後
在 redis 中就會多一個 key

redis new

假設我們要以 user_id 當 Key 的 Cache 資料只有 id , name , email , avatar 四個欄位
另一個以同樣以 user_id 最為 Key 的 Cache 資料只有 id , password 兩個欄位
還需要我們額外處理

又因為 redis-om 的 Model 是透過 pydantic 來實作
像是 email 綁定 EmailStr 、 birthday 綁定 date 會導致一些 Validation 的問題

email str error

因為 redis-om 綁定 pydantic 的關係
沒有辦法使用的這麼彈性

觀察 `redis-om` 結果

觀察 redis-om 的結果可以發現
是將 modulename + classname + pk 當作 key
如下所示：

redis hash model
( HashModel 生成的 Key 格式 )

redis json model
( JsonModel 生成的 Key 格式 )

以 `redis-py` 和自定義 Model 實作

既然使用 redis-om 會有額外的問題
也須要額外處理

那不如我們就直接使用 redis-py 來實作
同樣也是使用 decorator 來實作

cache decorator 設計

Cache 主要分成 3 個操作

get
update
delete

並使用 Redis 內建的 Hash Type 來實作

redis_connection.hgetall( key_name ) : 回傳 key_name 下所有的 key-value pair ( dict )
redis_connection.hset( key_name , mapping ) : 將 mapping 中的 key-value pair 寫入 key_name 中 ( 可以當作新增或更新 )

又考慮到複用性 ，我們將 perfix 和 key 作為參數傳入
讓能夠以 prefix + kargs[key] 作為 Key

所以要被 Wrap 起來的 function 必須要有 **kwargs 來傳參數

database/redis_cache.py

def generic_cache_get(prefix:str,key:str,cls:object):
    '''
    prefix: namspace for redis key ( such as `user` 、`item` 、`article` )
    key: **parameter name** in caller function ( such as `user_id` 、`email` 、`item_id` )
    cls: **response schema** in caller function ( such as `UserSchema.UserRead` 、`UserSchema.UserId` 、`ItemSchema.ItemRead` )
    '''

    rc = redis.Redis(connection_pool=redis_pool)

    def inner(func):
        async def wrapper(*args, **kwargs):
            # TODO ...
        return wrapper
    return inner

如果有 cache_key 還需要檢查目前 Cache 的結果是否有所有 Schema 包含的 Field
如果有缺少 Field 話
也需要先去 DB 中取得資料後再更新！

database/redis_cache.py

    #...
    # 在呼叫被包裝的 function 時，要以 func(key=value) 的方式傳入參數
    value_key = kwargs.get(key) 
    if not value_key:
        return await func(*args, **kwargs)
    
    # 組合成給 redis 的 key
    cache_key = f"{prefix}:{value_key}"

    # 檢查是否有 cache
    try:
        redis_result:dict = rc.hgetall(cache_key)

        # 即使有結果 還需要檢查是否有所有的 Field
        if check_has_all_keys(redis_result,cls): # cache hit !
            return cls(**redis_result) 
    except:
        pass

    sql_result = await func(*args, **kwargs) 
    if not sql_result:
        return None
    
    rc.hset(cache_key, mapping=sql_query_row_to_dict(sql_result))
    return sql_result

我們這邊直接用 exception 來判斷是否有 cache hit
而不用 redis_connect.exists 來檢查來加速

使用 cache decorator

在 CRUD 中，我們只需要加上 @generic_cache_get 來使用
crud/users.py

# ...
    @generic_cache_get(prefix="user",key="user_id",cls=UserSchema.UserRead)
        async def get_user_infor_by_id(self,user_id:int ,db_session:AsyncSession) -> UserSchema.UserInfor:
            # ...

    @generic_cache_get(prefix="user",key="user_id",cls=UserSchema.UserId)
    async def get_user_id_by_id(self,user_id:int ,db_session:AsyncSession=None) -> UserSchema.UserId:
        # ...
    

    @generic_cache_get(prefix="user",key="email",cls=UserSchema.UserInDB)
    async def get_user_in_db(self,email:str , db_session:AsyncSession=None) -> UserSchema.UserInDB :
        # ...
# ...

要注意的是
當我們在 call 這些 CRUD function 時，要以 func(key=value) 的方式傳入參數

api/users.py

# ...

    # ...
    async def get_user_infor_by_id(user_id: int):
        # 這邊要以 `user_id=user_id` 的方式傳入
        user = await UserCrud.get_user_infor_by_id(user_id=user_id)

api/auth.py

async def login(form_data: login_form_schema):    # ...

    # 要以 `email=form_data.username` 的方式傳入
    user_in_db:UserInDB = await UserCrud.get_user_in_db(email=form_data.username)